/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data.a2b.worker;

import cn.ac.iie.ulss.data.a2b.commons.RuntimeEnv;
import cn.ac.iie.ulss.data.a2b.rocketmq.RocketMQProducer;
import cn.ac.iie.ulss.data.a2b.server.Dataa2bStartUp;
import static cn.ac.iie.ulss.data.a2b.server.Dataa2bStartUp.protocol;
import java.io.ByteArrayOutputStream;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.LinkedList;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryEncoder;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.log4j.Logger;

/**
 *
 * @author Li Mingyang
 */
public class Worker implements Runnable {

    public static Logger log = Logger.getLogger(Worker.class.getName());
    private String topic;
    private String schemaName;
    private int batchSize;

    public void init() {
        topic = RuntimeEnv.getParamAsString(RuntimeEnv.DES_TOPIC);
        schemaName = RuntimeEnv.getParamAsString(RuntimeEnv.TABLENAME);
        batchSize = RuntimeEnv.getParamAsInteger(RuntimeEnv.BATCHSIZENUM);
    }

    @Override
    public void run(){
        long predictedSendcount = 0L;
        int predictedMessageSize = 0;

        long lastSendtime = System.currentTimeMillis();
        Schema docsSchema = protocol.getType("docs");
        Schema singleDocSchema = protocol.getType(this.schemaName.toLowerCase());

        GenericRecord docsRecord = new GenericData.Record(docsSchema);
        GenericArray docSet = new GenericData.Array(this.batchSize, docsSchema.getField("doc_set").schema());

        ByteArrayOutputStream docsBos = new ByteArrayOutputStream();
        BinaryEncoder docsEncoder = new EncoderFactory().binaryEncoder(docsBos, null);
        DatumWriter docsWriter = new GenericDatumWriter(docsSchema);

        ByteArrayOutputStream singledocBos = new ByteArrayOutputStream();
        BinaryEncoder singledocEncoder = new EncoderFactory().binaryEncoder(singledocBos, null);
        DatumWriter singledocWriter = new GenericDatumWriter(singleDocSchema);

        LinkedList<GenericRecord> innerBuffer = new LinkedList<GenericRecord>();
        Iterator<GenericRecord> it = innerBuffer.iterator();
        LinkedList<GenericRecord> tmp = new LinkedList<GenericRecord>();

        GenericRecord gr = null;
        try {
            while (true) {
                Dataa2bStartUp.a2bLinkedBlockingQueue.drainTo(innerBuffer, 300);
                /*处理过大消息体单独存放的buffer*/
                innerBuffer.addAll(tmp);
                tmp.clear();
                if (!innerBuffer.isEmpty()) {
                    for (it = innerBuffer.iterator(); it.hasNext();) {
                        predictedSendcount++;
                        gr = it.next();
                        singledocWriter.write(gr, singledocEncoder);
                        singledocEncoder.flush();
                        predictedMessageSize += singledocBos.size();
                        if (predictedMessageSize >= RuntimeEnv.getParamAsInteger(RuntimeEnv.MAX_MESSAGE_SIZE)) {
                            /*把不能发出的数据放入一个临时的buffer，下次处理*/
                            tmp.add(gr);

                            docsRecord.put("doc_schema_name", this.schemaName);
                            docsRecord.put("doc_set", docSet);
                            docsRecord.put("sign", "this is for sign");

                            docsWriter.write(docsRecord, docsEncoder);
                            docsEncoder.flush();
                            log.warn(" message size will over max size, original data size is " + (predictedMessageSize - singledocBos.size()) + ",and final data size is " + docsBos.size() + " for " + this.schemaName);
                            predictedSendcount--;
                            this.send(this.topic, docsBos.toByteArray(), predictedSendcount);

                            singledocBos.reset();
                            docsBos.reset();

                            predictedSendcount = 0;
                            predictedMessageSize = 0;
                            lastSendtime = System.currentTimeMillis();

                            docsRecord = new GenericData.Record(docsSchema);
                            docSet = new GenericData.Array(this.batchSize, docsSchema.getField("doc_set").schema());
                        } else if (docSet.size() >= this.batchSize - 1) {
                            docSet.add(ByteBuffer.wrap(singledocBos.toByteArray()));
                            docsRecord.put("doc_schema_name", this.schemaName);
                            docsRecord.put("doc_set", docSet);
                            docsRecord.put("sign", "this is for sign");

                            docsWriter.write(docsRecord, docsEncoder);
                            docsEncoder.flush();
                            this.send(this.topic, docsBos.toByteArray(), predictedSendcount);

                            singledocBos.reset();
                            docsBos.reset();

                            predictedSendcount = 0;
                            predictedMessageSize = 0;
                            lastSendtime = System.currentTimeMillis();

                            docsRecord = new GenericData.Record(docsSchema);
                            docSet = new GenericData.Array(this.batchSize, docsSchema.getField("doc_set").schema());
                        } else {
                            docSet.add(ByteBuffer.wrap(singledocBos.toByteArray()));
                            singledocBos.reset();
                        }
                    }
                    innerBuffer.clear();
                } else {
                    /*如果list是空的就sleep一下*/
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException ex) {
                    }
                    if (System.currentTimeMillis() - lastSendtime >= 5000) {
                        docsRecord.put("doc_schema_name", this.schemaName);
                        docsRecord.put("doc_set", docSet);
                        docsRecord.put("sign", "this is for sign");
                        docsWriter.write(docsRecord, docsEncoder);
                        docsEncoder.flush();
                        if (!docSet.isEmpty()) {
                            this.send(this.topic, docsBos.toByteArray(), predictedSendcount);
                        }

                        docsBos.reset();

                        predictedSendcount = 0;
                        predictedMessageSize = 0;
                        lastSendtime = System.currentTimeMillis();

                        docsRecord = new GenericData.Record(docsSchema);
                        docSet = new GenericData.Array(this.batchSize, docsSchema.getField("doc_set").schema());
                    }
                }
            }
        } catch (Exception ex) {
            log.error(ex.getMessage(), ex);
        }
    }

    private void send(String topic, byte[] pData, long count) throws Exception {
        RocketMQProducer.sendMessage(topic, pData, count);
    }

}
